package com.atguigu.networkflow.analysis;

import com.atguigu.networkflow.analysis.bean.PageViewCount;
import com.atguigu.networkflow.analysis.bean.UserBehavior;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.net.URL;

public class UvWithBloomFilter {
    public static void main(String[] args) throws Exception {
        //1.创建执行环节
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //2.读取数据
        URL resource = UniqueVisitor.class.getResource("/UserBehavior.csv");
        DataStream<String> inputStream = env.readTextFile(resource.getPath());

        //3.转换成pojo，分配时间戳和watermark
        DataStream<UserBehavior> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(new Long(fields[0]), new Long(fields[1]), new Integer(fields[2]), fields[3], new Long(fields[4]));
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
            @Override
            public long extractAscendingTimestamp(UserBehavior userBehavior) {
                return userBehavior.getTimestamps() * 1000L;
            }
        });

        //开窗统计uv值
        SingleOutputStreamOperator<PageViewCount> uvStream = dataStream
                .filter(data -> "pv".equals(data.getBehavior()))
                .timeWindowAll(Time.hours(1))
                .trigger(new MyTrigger())
                .process(new UvCountResultWithBloomFilter());

        uvStream.print();

        env.execute("uv with bloom filter job");
    }

    public static class MyTrigger extends Trigger<UserBehavior, TimeWindow> {

        //来了一条数据时应该执行的操作
        @Override
        public TriggerResult onElement(UserBehavior userBehavior, long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
            //每一条数据来到，直接触发窗口计算，并且清空数据
            return TriggerResult.FIRE_AND_PURGE;
        }

        //处理中时执行操作
        @Override
        public TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
            return TriggerResult.CONTINUE;
        }

        //事件事件改编执行操作
        @Override
        public TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
            return TriggerResult.CONTINUE;
        }

        //清除临时状态
        @Override
        public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {

        }
    }

    //实际情况则是引入第三方的google的 bloom filter
    //自定义一个布隆过滤器
    public static class MyBloomFilter {
        //定义位图的大小,一般需要定义为2的整次幂,最好是8的整次幂
        private Integer cap;

        public MyBloomFilter(Integer cap) {
            this.cap = cap;
        }

        //实现一个hash函数
        public Long hashCode(String value, Integer seed) {
            Long result = 0L;
            for (int i = 0; i < value.length(); i++) {
                result = result * seed + value.charAt(i);
            }

            return result & (cap - 1);
        }
    }

    //实现自定义的处理函数
    public static class UvCountResultWithBloomFilter extends ProcessAllWindowFunction<UserBehavior, PageViewCount, TimeWindow> {

        //定义jedis连接和bloom filter
        Jedis jedis;
        MyBloomFilter myBloomFilter;

        @Override
        public void open(Configuration parameters) throws Exception {
            jedis = new Jedis("localhost",6379);
            //要处理一亿个数据，64MB大小的位图
            myBloomFilter = new MyBloomFilter(1 << 29);
        }

        @Override
        public void process(Context context, Iterable<UserBehavior> elements, Collector<PageViewCount> out) throws Exception {
            //将位图和窗口的count值全部存入redis,用window end作为key
            Long windowEnd = context.window().getEnd();
            String bitMapKey = windowEnd.toString();
            //把count值存成一张hash表
            String countHashName = "uv_count";
            String countKey = windowEnd.toString();

            //1.取当前的user id
            Long userId =  elements.iterator().next().getUserId();
            //2.计算位图中的偏移量
            Long offset = myBloomFilter.hashCode(userId.toString(),61);
            //3.用redis的getbit命令判断对应位置的值
            Boolean isExist = jedis.getbit(bitMapKey,offset);

            if(!isExist){
                //4.不存在对应位图位置为1
                jedis.setbit(bitMapKey,offset,true);
                //更新redis中保存的count值
                Long uvCount = 0L;
                String uvCountString = jedis.hget(countHashName, countKey);
                if(uvCountString!=null && !"".equals(uvCountString)){
                    uvCount = Long.valueOf(uvCountString);
                }
                jedis.hset(countHashName,countKey,String.valueOf(uvCount+1));

                out.collect(new PageViewCount("uv",windowEnd,uvCount+1));
            }

        }

        @Override
        public void close() throws Exception {
            jedis.close();
        }
    }
}
